package com.pipe.connector.actor;

import com.pipe.connector.RowDataListener;
import com.pipe.exception.ProtocolCodecException;
import com.pipe.exception.SessionException;
import com.pipe.protocol.BasePackage;
import com.pipe.protocol.events.table.TableMapEvent;
import com.pipe.protocol.login.ComBinlogDumpPackage;
import com.pipe.protocol.login.ComRegisterSlavePackage;
import com.pipe.protocol.login.HandshakeResponsePacket;
import com.pipe.protocol.query.ColumnDefinition41Package;
import com.pipe.protocol.query.ComQueryPackage;
import com.pipe.protocol.query.ResultsetRowPacakge;
import com.pipe.protocol.query.TableMetaInfo;
import com.pipe.util.MyBytebufUtil;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author Kiven
 */
public class SessionStatusActor {
    private static Logger LOG = LoggerFactory.getLogger(SessionStatusActor.class);

    private String databaseName;
    private int binlogPos;
    private String binlogFile;
    private TableMapEvent tableMapEvent;
    private SessionStatus sessionStatus = SessionStatus.UNLOGIN;

    private List<TableMetaInfo> tableList = new ArrayList<>(20);
    private int currentDescTableIndex;
    private Map<String, TableMetaInfo> tableMap = new HashMap<>(20);
    private DoLoginActor doLoginActor;
    private LoggedInActor loggedInActor = new LoggedInActor();
    private RegisteredActor registeredActor = new RegisteredActor();
    private QueryResultActor queryResultActor = new QueryResultActor();
    private BinlogEventActor binlogEventActor;

    public SessionStatusActor(String databaseName, String userName, String password, int binlogPos, String binlogFile, RowDataListener listener) {
        this.databaseName = databaseName;
        this.binlogPos = binlogPos;
        this.binlogFile = binlogFile;
        this.doLoginActor = new DoLoginActor(userName, password);
        this.binlogEventActor = new BinlogEventActor(listener);
    }

    public synchronized void handler(BasePackage basePackage, ChannelHandlerContext ctx) throws ProtocolCodecException, SessionException {
        if (sessionStatus == SessionStatus.UNLOGIN) {
            HandshakeResponsePacket handshakeResponsePacket = doLoginActor.handler(databaseName, basePackage);
            response(ctx, basePackage, handshakeResponsePacket.toBytebuf(), false);
            LOG.info("[LOGGING] - HandshakeResponsePacket sent!");
            sessionStatus = SessionStatus.LOGGING;
            return;
        }
        if (sessionStatus == SessionStatus.LOGGING) {
            loggedInActor.handler(basePackage);
            response(ctx, basePackage, ComRegisterSlavePackage.build(1).toBytebuf(), true);
            LOG.info("[REGISTERING] - ComRegisterSlavePackage sent!");
            sessionStatus = SessionStatus.REGISTERING;
            return;
        }
        if (sessionStatus == SessionStatus.REGISTERING) {
            registeredActor.handler(basePackage);
            response(ctx, basePackage, new ComQueryPackage("show tables from " + this.databaseName).toBytebuf(), true);
            LOG.info("[QUERYING] - Query string : [" + "show tables from " + this.databaseName + "] sent!");
            sessionStatus = SessionStatus.SHOW_TABLES;
            return;
        }
        if (sessionStatus == SessionStatus.SHOW_TABLES) {
            TableMetaInfo metaInfo = queryResultActor.handler(basePackage);
            if (metaInfo != null) {
                for (ResultsetRowPacakge pacakge : metaInfo.getResultList()) {
                    String tableName = pacakge.getValue().get(0);
                    tableList.add(new TableMetaInfo(this.databaseName, tableName));
                }
                moreTableNeedDesc(ctx, basePackage);
                sessionStatus = SessionStatus.DESC_TABLES;
            }
            return;
        }
        if (sessionStatus == SessionStatus.DESC_TABLES) {
            TableMetaInfo metaInfo = queryResultActor.handler(basePackage);
            if (metaInfo != null) {
                TableMetaInfo tableMetaInfo = this.tableList.get(this.currentDescTableIndex);
                tableMetaInfo.build(metaInfo.getColumnDefinitionList(), metaInfo.getResultList());
                tableMap.put(tableMetaInfo.getTableName(), tableMetaInfo);
                this.currentDescTableIndex++;
                boolean hasMore = moreTableNeedDesc(ctx, basePackage);
                if (!hasMore) {
                    LOG.info("[QUERYING] - no more table needs to be desc!");
                    response(ctx, basePackage, ComBinlogDumpPackage.build(this.binlogPos, (short) 0, 11, this.binlogFile).toBytebuf(), true);
                    LOG.info("[DUMPING] - ComBinlogDumpPackage sent! Pls waiting for the new data!");
                    sessionStatus = SessionStatus.DUMP_REQUESTED;
                }
            }
            return;
        }

        if (sessionStatus == SessionStatus.DUMP_REQUESTED) {
            TableMapEvent tableMapEvent = binlogEventActor.handler(basePackage, this.getTableMapEvent(), getTableMap());
            if (tableMapEvent != null) {
                this.setTableMapEvent(tableMapEvent);
            }
        }
    }

    boolean moreTableNeedDesc(ChannelHandlerContext ctx, BasePackage basePackage) {
        if (this.currentDescTableIndex < this.tableList.size()) {
            TableMetaInfo tableMetaInfo = this.tableList.get(this.currentDescTableIndex);
            response(ctx, basePackage, new ComQueryPackage("desc " + this.databaseName + "." + tableMetaInfo.getTableName()).toBytebuf(), true);
            LOG.info("[QUERYING] - Query string : [" + "desc " + this.databaseName + "." + tableMetaInfo.getTableName() + "] sent!");
            return true;
        }
        return false;
    }

    void response(ChannelHandlerContext ctx, BasePackage basePackage, ByteBuf responseArr, boolean newSession) {
        if (newSession) {
            basePackage.resetPayloadAndSession(responseArr);
        } else {
            basePackage.resetPayload(responseArr);
        }
        ctx.writeAndFlush(basePackage.toBytebuf());
    }

    public void setTableMapEvent(TableMapEvent tableMapEvent) {
        this.tableMapEvent = tableMapEvent;
    }

    public TableMapEvent getTableMapEvent() {
        return tableMapEvent;
    }

    public Map<String, TableMetaInfo> getTableMap() {
        return tableMap;
    }
}